package com.jzl.canal_consumer;


import com.jzl.canal_consumer.config.CustomConfig;
import com.jzl.canal_consumer.etl.et.rocketmq.consumer.CanalRocketMqConsumer;
import com.jzl.canal_consumer.etl.et.rocketmq.consumer.CanalRocketMqConsumerFactory;
import com.jzl.canal_consumer.service.MacService;
import com.jzl.util.LoggerUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import com.jzl.canal_consumer.bean.OpsDataSource;

import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

/**
 * 消费启动器
 *
 * @author ZhenWuWang
 */
@Component
public class ConsumerStarter
{
    private final MacService macService;

    private final CustomConfig customConfig;

    private Map<OpsDataSource, List<CanalRocketMqConsumer>> consumerMap;

    private final ScheduledExecutorService executorService;

    private final CanalRocketMqConsumerFactory consumerFactory;

    @Autowired
    public ConsumerStarter(MacService macService, CustomConfig customConfig, CanalRocketMqConsumerFactory consumerFactory)
    {
        this.macService = macService;
        this.customConfig = customConfig;
        this.executorService = Executors.newScheduledThreadPool(customConfig.getConsumerMaxCount());
        this.consumerFactory = consumerFactory;
    }

    void start()
    {
        //获取同步源信息列表
        List<OpsDataSource> macList = macService.listOpsDataSource();
        if (macList == null || macList.isEmpty())
        {
            LoggerUtil.info("## mac 列表为空！");
            return;
        }
        //获取Rokcetmq GroupSuffixes
        //每组的数据源推送的消息通道 group=mac+GroupSuffixes;topic = mac+GroupSuffixes
//        String groupSuffixes = customConfig.getGroupSuffixes();
//        final List<String> groupSuffixList = new ArrayList<>();
//        if (!StringUtils.isEmpty(groupSuffixList))
//        {
//            String[] ss = groupSuffixes.split(",");
//            if (ss.length > 0)
//            {
//                groupSuffixList.addAll(Arrays.asList(ss));
//            }
//        }
        //获取当前消息解析类型
        boolean flatMessage = customConfig.getParseType() != 0;
        LoggerUtil.info("## 当前数据解析方式:" + (!flatMessage ? "序列化" : "json(FlatMessage)"));
        //
        String nameServers = customConfig.getNameServers();
        consumerMap = new HashMap<>(macList.size());
        int consumerCount = 0;
        List<String> topicSuffixList = new ArrayList<>();
        String topicSuffixes = customConfig.getTopicSuffixes();
        if (!StringUtils.isEmpty(topicSuffixes))
        {
            String[] ss = topicSuffixes.split(",");
            if (ss.length > 0)
            {
                topicSuffixList.addAll(Arrays.asList(ss));
            }
        }
        for (OpsDataSource opsDataSource : macList)
        {
//            List<CanalRocketMqConsumer> consumers = consumerFactory.canalRocketMqConsumerList(groupSuffixList,
//                    nameServers, opsDataSource.getMac(), opsDataSource.getSchemaName(), flatMessage);
//            if (consumers != null && consumers.size() > 0)
//            {
//                consumerMap.put(opsDataSource, consumers);
//                consumerCount += consumers.size();
//            }
            String groupName = opsDataSource.getMac() + "_OPS";
            String[] topics = topicSuffixList.stream().map(s -> opsDataSource.getMac() + s).toArray(String[]::new);
            CanalRocketMqConsumer consumer = consumerFactory.createCanalRocketMqConsumer(nameServers, groupName,
                    topics, opsDataSource.getSchemaName(), flatMessage);
            if (consumer != null)
            {
                consumerMap.put(opsDataSource, Collections.singletonList(consumer));
                consumerCount++;
            }
        }

        if (consumerCount > customConfig.getConsumerMaxCount())
        {
            throw new IllegalArgumentException(String.format("## canal消息消费者数量超过了最大允许数量%S", customConfig.getConsumerMaxCount()));
        } else if (consumerCount == 0)
        {
            throw new RuntimeException("CanalRocketMqConsumer 数量为0");
        }

        consumerMap.values().forEach(consumers -> consumers.forEach(consumer ->
        {
            executorService.execute(consumer::start);
        }));

        Runtime.getRuntime().addShutdownHook(new Thread(() ->
        {
            try
            {
                LoggerUtil.info("## Stop all the rocketmq consumer");
                executorService.shutdown();
                consumerMap.clear();
            } catch (Throwable e)
            {
                LoggerUtil.warn("## Something goes wrong when stopping rocketmq consumer:", e);
            } finally
            {
                LoggerUtil.info("## Rocketmq consumer is down.");
            }
        }));
    }
}
